package org.sn.jdish.spark.operator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;
import java.util.Arrays;

/**
 * 单词统计排序
 * 
 * @author snzigod@hotmail.com
 *
 */
public final class WordCount {

	public static void main(String[] args) throws Exception {
		/**
		 * 第1步：创建SparkConf对象。 Master设置程序链接的Spark集群URL（设置为local，代表Spark程序在本地运行）。
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求。
		 */
		SparkConf sf = new SparkConf().setAppName("sparkOperator").setMaster("local");

		/**
		 * 第2步：创建SparkContext对象。
		 * SparkContext是通往集群的唯一通道，SparkContext初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler、TaskScheduler、SchedulerBackend
		 * 同时还会负责Spark程序往Master注册程序等。
		 */
		JavaSparkContext sc = new JavaSparkContext(sf);// 底层是scala的sparkcontext。

		wordCount(sc);

		sc.close();
	}

	public static void wordCount(JavaSparkContext sc) {
		/**
		 * 第3步：根据具体的数据来源（HDFS、HBase、Local FS、DB、S3等）创建RDD。
		 * JavaRDD创建的三种方式：外部数据源（如HDFS）、Scala集合、其它RDD操作。
		 * JavaRDD将数据划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴。
		 */
		JavaRDD<String> text = sc.textFile("data/wordCount.txt");

		/**
		 * 第4步：对初始的JavaRDD进行Transformation级别处理（如map、filter等高阶函数）来进行具体的数据计算。
		 */
		/**
		 * 第4.1步：将每一行的字符串拆分成单个的单词。
		 */
		JavaRDD<String> words = text.flatMap(new FlatMapFunction<String, String>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<String> call(String line) throws Exception {
				return Arrays.asList(line.split(" "));
			}
		});
		/**
		 * 第4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是word => (word, 1)
		 */
		JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, Integer> call(String word) throws Exception {
				return new Tuple2<String, Integer>(word, 1);
			}
		});
		/**
		 * 第4.3步：在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
		 */
		JavaPairRDD<String, Integer> wordsCount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
			private static final long serialVersionUID = 1L;

			@Override
			public Integer call(Integer value1, Integer value2) throws Exception {
				return value1 + value2;
			}
		});

		wordsCount.foreach(new VoidFunction<Tuple2<String, Integer>>() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Tuple2<String, Integer> pairs) throws Exception {
				System.out.println(pairs._1 + " : " + pairs._2);
			}
		});

		// 键值对互换
		JavaPairRDD<Integer, String> temp = wordsCount
				.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
					private static final long serialVersionUID = 1L;

					@Override
					public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {
						return new Tuple2<Integer, String>(tuple._2, tuple._1);
					}
				});

		temp.foreach(new VoidFunction<Tuple2<Integer, String>>() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Tuple2<Integer, String> pairs) throws Exception {
				System.out.println(pairs._1 + " : " + pairs._2);
			}
		});

		// 排序
		JavaPairRDD<String, Integer> sorted = temp.sortByKey(false)
				.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
					private static final long serialVersionUID = 1L;

					@Override
					public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple) throws Exception {
						return new Tuple2<String, Integer>(tuple._2, tuple._1);
					}
				});

		sorted.foreach(new VoidFunction<Tuple2<String, Integer>>() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(Tuple2<String, Integer> pair) throws Exception {
				System.out.println(pair._1 + " : " + pair._2);
			}
		});

		for (Tuple2<String, Integer> pair : sorted.collect()) {
			System.out.println(pair._1 + " : " + pair._2);
		}
	}

}